package ex.datastream.functions.dataStream;

import ex.datastream.functions.richFunction.RichCoFlatMapFunc;
import ex.vo.Card;
import ex.vo.Person;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


/**
 * 将2个不同类型的 dataStream 合并成一个新的 dataStream
 *
 * https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/dev/datastream/operators/overview/#comap-coflatmap
 */
public class ConnectFuncTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        DataStream<Person> ds1 = env.fromElements(
                new Person("Fred", 35),
                new Person("Wilma", 35),
                new Person("Pebbles", 2)).keyBy(p->p.name);

        DataStream<Card> ds2 = env.fromElements(
                new Card("Fred", 123),
                new Card("Wilma", 456),
                new Card("Pebbles", 789)).keyBy(p->p.name);

        ConnectedStreams<Person, Card> connectedStreams = ds1.connect(ds2);

        connectedStreams.flatMap(new RichCoFlatMapFunc()).print();

        env.execute();
    }
//    PersonCard{person=Person{name='Fred', age=35}, card=Card{name='Fred', no=123}}
//    PersonCard{person=Person{name='Wilma', age=35}, card=Card{name='Wilma', no=456}}
//    PersonCard{person=Person{name='Pebbles', age=2}, card=Card{name='Pebbles', no=789}}
}
